package com.inji.spark.biz.service.kafka;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.regex.Pattern;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Lists;
import com.inji.spark.biz.dto.SparkApiLogReq;
import com.inji.spark.biz.listener.SparkContextInit;
import com.inji.spark.biz.util.JsonMapper;

/**
 * @author liukz
 *
 */
public class KafkaReader {
	static final Pattern SPACE = Pattern.compile(" ");
	private static Logger logger = LoggerFactory.getLogger(KafkaReader.class);
	
	public static void main(String[] args) {
		initInjiKafkaReader();
	}

	public static void initInjiKafkaReader() {

		// 每个话题的分片数
		int numThreads = 2;
		SparkConf sparkConf = new SparkConf().setAppName("KafkaWordCount")
				.setMaster("local[2]");
		JavaStreamingContext jssc = new JavaStreamingContext(sparkConf,
				new Duration(10000));

		Map<String, Object> kafkaParams = new HashMap<String, Object>();
		kafkaParams.put("bootstrap.servers", "172.16.201.240:9092");
		kafkaParams.put("key.serializer", StringSerializer.class);
		kafkaParams.put("key.deserializer", StringDeserializer.class);
		kafkaParams.put("value.deserializer", StringDeserializer.class);
		kafkaParams.put("group.id", "use_a_separate_group_id_for_each_stream");
		kafkaParams.put("auto.offset.reset", "latest");
		kafkaParams.put("enable.auto.commit", false);

		Collection<String> topics = Arrays.asList("test", "test2");

		final JavaInputDStream<ConsumerRecord<String, String>> stream = KafkaUtils
				.createDirectStream(jssc,
						LocationStrategies.PreferConsistent(),
						ConsumerStrategies.<String, String> Subscribe(topics,
								kafkaParams));
		JavaDStream<String> words = stream
				.flatMap(new FlatMapFunction<ConsumerRecord<String, String>, String>() {
					public Iterator<String> call(
							ConsumerRecord<String, String> t) throws Exception {
						SparkApiLogReq req = JsonMapper.buildNonDefaultMapper().fromJson(t.value(), SparkApiLogReq.class);
						SparkContextInit sparkContextInit = new SparkContextInit();
						sparkContextInit.getApiRespContent(req);
						return Lists.newArrayList(SPACE.split(t.value()))
								.iterator();
					}
				});
	

		words.print();

		try {
			jssc.start();
			jssc.awaitTermination();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

	}
}